/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.io.IOException;

import org.singularity.io.AbstractTransitPipe;
import org.singularity.io.CloseablePipe;
import org.singularity.io.DME;
import org.singularity.mapred.MapFun;
import org.singularity.mapred.VDME;

/**
 * @author vjache
 *
 */
public class MapperPipe<VT, KT1, VT1> extends AbstractTransitPipe<DME<VT>,VDME<KT1, VT1>> {
	
	static final String MAPPER_ID_ANNOT = "mpr.id";

	private final MapFun<VT, KT1, VT1> _fun;

	private final Object _id;
	
	private class AnnotatorPipe extends AbstractTransitPipe<VDME<KT1, VT1>,VDME<KT1, VT1>>
	{

		AnnotatorPipe(CloseablePipe<VDME<KT1, VT1>> aUnder)
		{
			super(aUnder);
		}

		@Override
		public void push(VDME<KT1, VT1> aItem) throws IOException {
			aItem.putAnnotation(MAPPER_ID_ANNOT, _id);
			this._next.push(aItem);
		}
		
	}

	public MapperPipe(
			Object aMprId,
			CloseablePipe<VDME<KT1, VT1>> aNextOne, 
			MapFun<VT, KT1, VT1> aMapFun) {
		super(aNextOne);
		_id = aMprId;
		_fun = aMapFun;
	}
	
	@Override
	protected CloseablePipe<VDME<KT1, VT1>> onSetUpPipe(CloseablePipe<VDME<KT1, VT1>> aNextOne) {
		return new AnnotatorPipe(aNextOne);
	}

	@Override
	public void push(DME<VT> aItem) throws IOException {
		try {
			_fun.map(aItem, _next);
		} catch (Exception e) {
			throw new IOException(e);
		}
	}	
}
